package mainfuncs

import (
	"sync"

	"github.com/Shopify/sarama"
	"github.com/astaxie/beego/logs"
)

//AppKafkaClient 保存kafka消费方信息
var (
	AppKafkaClient *KafkaClient
)

//KafkaClient 保存kafka消费方信息
type KafkaClient struct {
	client sarama.Consumer
	addr   string
	topic  string
	wg     sync.WaitGroup
}

//InitKafka 初始化消息消费方，topic应该放在etcd动态读取
func InitKafka(addr string, topic string) (err error) {

	AppKafkaClient = &KafkaClient{}

	consumer, err := sarama.NewConsumer([]string{addr}, nil)
	if err != nil {
		logs.Error("init kafka failed, err:%v", err)
		return
	}

	AppKafkaClient.client = consumer
	AppKafkaClient.addr = addr
	AppKafkaClient.topic = topic
	return
}
